package com.ruyuan.engine.timer;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

/**
 * 定时场景的案例：满足第一套规则触发第二套定时规则
 * K2523KFDSK,oper1,1643203271000
 * K2523KFDSK,oper5,1643203272000
 * K2523KFDSK,oper3,1643203273000
 * K2523KFDSK,oper4,1643203275000
 * K2523KFDSK,oper4,1643203276000
 * K2523KFDSK,oper4,1643203276001 ：触发
 * K2523KFDSK,oper4,1643203278000
 * K2523KFDSK,oper4,1643203278001 ：允许延迟
 *
 *  满足规则：
 * K2523KFDSK,oper1,1643203271000
 * K2523KFDSK,oper2,1643203272000
 * K2523KFDSK,oper3,1643203273000
 * K2523KFDSK,oper4,1643203275000
 * K2523KFDSK,oper4,1643203276000
 * K2523KFDSK,oper4,1643203276001 ：应该触发时间
 * K2523KFDSK,oper4,1643203278000
 * K2523KFDSK,oper4,1643203278001 ：实际触发时间（允许延迟）
 * .....
 * */
public class RegisterOnTimer {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        environment.setParallelism(1);
        DataStreamSource<String> source = environment.socketTextStream("hadoop5", 9991);
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = source.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofMillis(2000)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {

            /**
             * 基于EventTime处理
             * */
            @Override
            public long extractTimestamp(String log, long timeStamp) {
                return Long.parseLong(log.split(",")[2]);
            }
        }));

        SingleOutputStreamOperator<String> process = stringSingleOutputStreamOperator.keyBy(log -> log.split(",")[0]).process(new KeyedProcessFunction<String, String, String>() {

            ListState<Tuple2<String,Long>> listState;

            @Override
            public void open(Configuration parameters) throws Exception {
                listState = getRuntimeContext().getListState(new ListStateDescriptor<Tuple2<String, Long>>("listState", TypeInformation.of(new TypeHint<Tuple2<String, Long>>() {
                })));
            }

            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                String[] logs = value.split(",");
                String eventTime = logs[2];
                String oper = logs[1];
                if (oper.equals("oper1")) {
                    System.out.println("做了oper1,注册定时器，在指定的时间是否满足后续事件");
                    ctx.timerService().registerEventTimeTimer(Long.parseLong(eventTime) + 5000);
                }
                /**
                 * 定时规则：当客户做了oper1，那么在接下来的10秒需要做oper2，
                 * */
                if (oper.equals("oper2")) {
                    //使用State进行存储
                    listState.add(Tuple2.of(oper,Long.parseLong(eventTime)));
                }

            }

            /**
             * 事件的定时器
             * */
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                boolean flag = false;
                System.out.println("定时器被触发了，触发的事件时间是："+timestamp);
                Iterator<Tuple2<String, Long>> iterator = listState.get().iterator();
                /**
                 * 思考：
                 *      用户1  oper1  监听 5秒
                 *              5秒以内又做了 oper1 再次监听
                 *      用户2  oper1  监听 5秒
                 *
                 *      数据都是存在listState，用户所有的操作的事件，
                 *      比如用户1  oper1  监听 5秒 触发了，就需要删除掉 oper1对应的oper2在缓存中的数据。
                 *      时间：oper2的时间必须要小于当前窗口触发的时间.
                 * */
                while (iterator.hasNext()) {
                    Tuple2<String, Long> oper2 = iterator.next();
                    if (oper2.f1 < timestamp) {
                        //在指定的时间内做了该操作
                        flag = true;
                        iterator.remove();
                    }
                }
                if (flag) out.collect(ctx.getCurrentKey()+ " 在触发了oper1事件后的指定时间内，做了oper2");
            }
        });
        process.print();
        environment.execute("RegisterOnTimer");
    }
}
